package edu.npu.GraphIndex;

import java.io.IOException;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
//在标签中对于边界点作出标记
public class BoundaryDetection {
	public static final String SEPARATOR = "\t";
	public static class BoundaryDetectionMapper extends Mapper<Text, Text, Text, Text> {
		@Override
		protected void map(Text key, Text value, Context context)
				throws IOException, InterruptedException {
			String line = value.toString();
			if(line.contains("#")) {//表示是边
				String[] splits = line.split(SEPARATOR);
				String[] kPart, vPart;
				kPart = splits[0].split("#");
				vPart = splits[1].split("#");
				context.write(new Text(kPart[0]), new Text("#" + kPart[1]));//标记下所属的part：<key, #part>
				context.write(new Text(vPart[0]), new Text("#" + vPart[1]));
			} else {//表示是标签
				String[] splits = line.split(SEPARATOR);
				context.write(new Text(splits[0]), new Text(splits[1]));
			}
		}
	}
	public static class BoundaryDetectionReducer extends Reducer<Text, Text, Text, Text> {
		@Override
		protected void reduce(Text key, Iterable<Text> values, Context context)
				throws IOException, InterruptedException {
			boolean isBoundary = false;
			int part = 0;
			Text label = null;
			for(Text value : values) {
				if(value.toString().contains("#")) {
					part = Integer.parseInt(value.toString().substring(1));
					isBoundary = true;
					if(label != null)
						break;
				} else {
					label = new Text(value);
				}
			}
			if(isBoundary) {
				context.write(new Text(part + "#" + key), label);//如果是边界的话标记一下是哪个部分的：<part#node,label>
			} else {
				context.write(key, label);
			}
		}
	}
	public static void main(String[] args) {
		
	}
}
